package com.qy566.gps.pulsar.producer;

import com.qy566.gps.common.pulsar.constant.Serialization;
import com.qy566.gps.common.pulsar.producer.ProducerFactory;
import org.apache.pulsar.client.admin.PulsarAdmin;
import org.apache.pulsar.client.admin.PulsarAdminException;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import java.util.HashMap;
import java.util.List;
import java.util.Map;

/**
 * @author wangnaiqi
 * @since 2024/1/17
 */
@Configuration
public class ProducerConfiguration {

    @Autowired
    private PulsarAdmin pulsarAdmin;

    @Value("${pulsar.tenant}")
    private String tenantName;

    @Bean
    public ProducerFactory producerFactory() throws PulsarAdminException {
        List<String> namespaces = pulsarAdmin.namespaces().getNamespaces(tenantName);
        Map<String, String> topicsMap = new HashMap<>();


        for (String namespaceNameUrl : namespaces){
            List<String> topicSub = pulsarAdmin.topics().getList(namespaceNameUrl);
            List<String> topicSubPartition = pulsarAdmin.topics().getPartitionedTopicList(namespaceNameUrl);

            for (String topic : topicSub) {
                String  namespaceName = namespaceNameUrl.substring(namespaceNameUrl.lastIndexOf("/") + 1);
                if(!topic.contains("partition")){
                    topicsMap.put(topic.substring(topic.lastIndexOf("/") + 1), namespaceName);
                }
            }

            for (String topic : topicSubPartition) {
                String  namespaceName = namespaceNameUrl.substring(namespaceNameUrl.lastIndexOf("/") + 1);
                topicsMap.put(topic.substring(topic.lastIndexOf("/") + 1), namespaceName);
            }

        }
        ProducerFactory producerFactory = new ProducerFactory();
        topicsMap.entrySet().stream()
                .forEach(entry -> producerFactory.addProducer(entry.getKey(), entry.getValue(), String.class, Serialization.STRING));

        return producerFactory;
    }
}
